package util

import (
	"context"
	"fmt"
	"log"
	"server-go/rpc/reward/internal/svc"
	"time"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

func ReckConsumerA(svcCtx *svc.ServiceContext, consumerA rocketmq.PushConsumer) {
	err := consumerA.Subscribe(svcCtx.Config.Rocketmq.TopicA, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			nowStr := time.Now().Format("2006-01-02 15:04:05")
			email := string(msg.Body)
			fmt.Printf("%s 读取到一条消息,消息内容: %s \n", nowStr, string(msg.Body))
			if err := SendCodeEmail(email, svcCtx); err != nil {
				log.Println("1111111111", err.Error())
				return consumer.Rollback, err
			}
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		log.Println("ConsumerA消费消息失败, " + err.Error())
	}
}
